package com.tbit.uqbike.client.service.event.impl;

import com.tbit.uqbike.client.dto.EventPutDTO;
import com.tbit.uqbike.client.service.event.EventHandleService;
import com.tbit.uqbike.client.util.aliyun.AliLogUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 处理事件上报 jvm + Scheduled 方案
 *
 * @author: qiu.longjie
 * @create: 2025-06-17
 */
@Service
@Slf4j
public class EventHandleServiceImpl implements EventHandleService {

    // 例如，每个事件1KB，10万个事件就是100MB。
    private static final int QUEUE_CAPACITY = 100_000;
    // 每次从队列中拉取的最大批次大小
    private static final int BATCH_SIZE = 5_000;
    // 单次 @Scheduled 任务执行允许的最大时间（毫秒），防止任务执行过久影响下一次调度
    private static final long MAX_PROCESSING_TIME_MILLIS = 4 * 1000; // 4秒，比5秒的调度周期短
    // 使用 LinkedBlockingQueue 作为并发安全的事件队列
    private final BlockingQueue<EventPutDTO> eventQueue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);

    @Resource
    private ThreadPoolExecutor orderTraceExecutor;

    @Override
    public void handle(EventPutDTO eventPutDTO) {
        if (eventPutDTO == null) {
            return;
        }
        try {
            // 使用 offer 而不是 add，因为 add 在队列满时会抛出 IllegalStateException
            // offer 在队列满时会返回 false，不会抛异常，可以进行流量控制
            boolean offered = eventQueue.offer(eventPutDTO); // 可以加超时 eventQueue.offer(eventPutDTO, 100, TimeUnit.MILLISECONDS);
            if (!offered) {
                log.warn("事件队列已满，事件 {} 被丢弃或未能入队。", eventPutDTO);
            }
        } catch (Exception e) {
            log.error("事件入队异常: {}", eventPutDTO, e);
        }
    }

    // 每5秒刷新一次，启动后5秒执行
    @Scheduled(fixedRate = 5 * 1000, initialDelay = 5 * 1000)
    public void processEvents() {
        if (eventQueue.isEmpty()) {
            return;
        }
        try {
            orderTraceExecutor.execute(this::doEvents);
        } catch (RejectedExecutionException e) {
            log.error("处理事件提交失败：线程池已满，任务被拒绝。当前队列大小: {}", eventQueue.size(), e);
            doEvents();
        } catch (Exception e) {
            log.error("处理事件埋点失败", e);
        }
    }

    private void doEvents() {
        long startTime = System.currentTimeMillis();
        // 记录本次任务总共处理的事件数
        int totalProcessedInThisRun = 0;
        // 只要队列不为空，且未超出单次任务的最大处理时间，就持续拉取和处理
        while (!eventQueue.isEmpty() && (System.currentTimeMillis() - startTime < MAX_PROCESSING_TIME_MILLIS)) {
            List<EventPutDTO> eventsToProcess = new ArrayList<>(BATCH_SIZE);
            // 将队列中最多 BATCH_SIZE 个元素转移到 eventsToProcess
            // drainTo 是原子操作，效率高
            int drainedCount = eventQueue.drainTo(eventsToProcess, BATCH_SIZE);
            if (CollectionUtils.isEmpty(eventsToProcess)) {
                // 如果 drainTo 没取到数据（可能是在 drainTo 瞬间队列变空），则退出循环
                break;
            }
            // 处理这批事件
            eventsToProcess.forEach(this::send);
            totalProcessedInThisRun += drainedCount;
        }
        long duration = System.currentTimeMillis() - startTime;
        log.info("【事件处理任务完成】本次处理 {} 个事件，耗时 {} ms，队列剩余大小: {}", totalProcessedInThisRun, duration, eventQueue.size());

        if (!eventQueue.isEmpty() && duration >= MAX_PROCESSING_TIME_MILLIS) {
            log.warn("任务因达到最大处理时间（{}ms）而提前结束，队列中仍有 {} 个事件。", MAX_PROCESSING_TIME_MILLIS, eventQueue.size());
        }
    }

    private void send(EventPutDTO eventPutDTO) {
        AliLogUtil.sendLog(eventPutDTO);
    }
}
